package com.bw.wjw.function;

import com.alibaba.fastjson.JSONObject;

import com.bw.wjw.DateTimeUtil;
import com.bw.wjw.bean.PageViewBean;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;

import java.util.HashMap;

/**
 * 将pageViewLog页面浏览日志数据封装到实体类PageViewBean对象
 * @author xuanyu
 * @date 2025/1/9
 */
public class PageViewBeanMapFunction extends RichMapFunction<JSONObject, PageViewBean> {


    //独立访客的优化
    // 1声明状态
    private ValueState<String> lastVisitDateState = null;

    private HashMap<String, String> areaDic=null;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 2初始化
        ValueStateDescriptor<String> lastVisitDate = new ValueStateDescriptor<>("lastVisitDate", String.class);

        StateTtlConfig build = StateTtlConfig.newBuilder(Time.days(1))
                .updateTtlOnCreateAndWrite()
                .neverReturnExpired()
                .build();

        lastVisitDate.enableTimeToLive(build);
        lastVisitDateState = getRuntimeContext().getState(lastVisitDate);

        areaDic=getAreaDic();

    }


    private HashMap<String, String> getAreaDic() {
        HashMap<String, String> map = new HashMap<>();
        map.put("北京","110000");
        map.put("天津","120000");
        map.put("山西","140000");
        map.put("内蒙古","150000");
        map.put("河北","130000");
        map.put("上海","310000");
        map.put("江苏","320000");
        map.put("浙江","330000");
        map.put("安徽","340000");
        map.put("福建","350000");
        map.put("江西","360000");
        map.put("山东","370000");
        map.put("台湾","710000");
        map.put("黑龙江","230000");
        map.put("吉林","220000");
        map.put("辽宁","210000");
        map.put("陕西","610000");
        map.put("甘肃","620000");
        map.put("青海","630000");
        map.put("宁夏","640000");
        map.put("新疆","650000");
        map.put("河南","410000");
        map.put("湖北","420000");
        map.put("湖南","430000");

        return map;
    }

    @Override
    public PageViewBean map(JSONObject value) throws Exception {
        // 提取字段值
        JSONObject commonJsonObject = value.getJSONObject("common");
        JSONObject pageJsonObject = value.getJSONObject("page");

        // todo 地区转换省份
        String province = commonJsonObject.getString("ar");

        // 会话数
        String lastPageIdValue = pageJsonObject.getString("last_page_id");
        Long sessionCount = (null == lastPageIdValue ? 1L : 0L);

        // 访客数
        long uvCount = 0L;
        // 上一次访问日期
        String lastVisitDate = lastVisitDateState.value();
        // 当前访问日期
        String visitDate = DateTimeUtil.convertLongToString(value.getLong("ts"), "yyyy-MM-dd");
        // 比较日期
        if (null == lastVisitDate || !lastVisitDate.equals(visitDate)) {
            uvCount = 1L;
            // 更新状态值
            lastVisitDateState.update(visitDate);
        }

        // 返回实体类对象
        return new PageViewBean(
                null,
                null,
                commonJsonObject.getString("ba"),
                commonJsonObject.getString("ch"),
                province,
                commonJsonObject.getString("is_new"),
                sessionCount,
                1L,
                pageJsonObject.getLong("during_time"),
                uvCount,
                value.getLong("ts")
        );
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
